package com.camemax.operators;

import com.camemax.pojo.DealCompareEvent;
import com.camemax.pojo.OrderPayEvent;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.co.CoProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

public class OrderCompareCoProcessFunction extends CoProcessFunction<OrderPayEvent
        , DealCompareEvent
        , Tuple2<OrderPayEvent, DealCompareEvent>> {

    // 定义状态
    ValueState<OrderPayEvent> orderEventState;
    ValueState<DealCompareEvent> dealEventState;

    @Override
    public void open(Configuration parameters) throws Exception {
        orderEventState = getRuntimeContext().getState(
                new ValueStateDescriptor<>("order-event",OrderPayEvent.class));
        dealEventState = getRuntimeContext().getState(
                new ValueStateDescriptor<>("deal-event",DealCompareEvent.class));
    }

    // connection已经将数据流按POJO类型分开
    // 执行processElement时，流入的数据流一定是OrderPayEvent（即平台的收据）对象，所以只需判断DealCompareEvent（第三方）对象是否已经存在
    @Override
    public void processElement1(OrderPayEvent value, Context ctx, Collector<Tuple2<OrderPayEvent, DealCompareEvent>> out) throws Exception {
        // 买家付完款后的平台收据
        if ( dealEventState.value() != null ){
            //
            out.collect(new Tuple2<>( orderEventState.value(), dealEventState.value()));
            orderEventState.clear();
            dealEventState.clear();
        }else{
            // 注册五秒后的时间，用于等待第三方支付平台的收据
            ctx.timerService().registerEventTimeTimer((value.getOrderTime()+5)*1000L);
            // 更新数据流状态为true， 表示【平台已经产生收据】
            orderEventState.update(value);
        }
    }

    @Override
    public void processElement2(DealCompareEvent value, Context ctx, Collector<Tuple2<OrderPayEvent, DealCompareEvent>> out) throws Exception {
        if ( orderEventState.value() != null ){
            out.collect(new Tuple2<>(orderEventState.value(), dealEventState.value()));
            orderEventState.clear();
            dealEventState.clear();
        }else{
            ctx.timerService().registerEventTimeTimer((ctx.timestamp() + 3 ) * 1000L);
            dealEventState.update(value);
        }
    }

    @Override
    public void onTimer(long timestamp, OnTimerContext ctx, Collector<Tuple2<OrderPayEvent, DealCompareEvent>> out) throws Exception {
        OutputTag<OrderPayEvent> orderTimeout = new OutputTag<OrderPayEvent>("order-timeout") {};
        OutputTag<DealCompareEvent> thirdPartyTimeout = new OutputTag<DealCompareEvent>("third-party-timeout"){};

        // 触发时间窗口计算的情况只能是平台或第三方支付中的其中一者为NULL
        if (orderEventState.value() != null){
            ctx.output(orderTimeout, orderEventState.value());
            orderEventState.clear();
        }
        if ( dealEventState.value() != null ){
            ctx.output(thirdPartyTimeout, dealEventState.value());
            dealEventState.clear();
        }
    }
}
